package com.youzhi.message.kafka;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class KafkaConfig {

    private String kafkaServer;

    public KafkaConfig(String kafkaServer) {
        this.kafkaServer = kafkaServer;
    }

    private Properties getConfig() {
        Properties prop = new Properties();
        prop.put("bootstrap.servers", this.kafkaServer);
        prop.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");

        prop.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        prop.put("group.id", "con-1");
        prop.put("auto.offset.reset", "latest");
        //自动提交偏移量
        prop.put("auto.commit.intervals.ms", "true");
        //自动提交时间
        prop.put("auto.commit.interval.ms", "1000");

        return prop;
    }


    public KafkaConsumer<String, String> getKafkaConsumer() {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(getConfig());
        consumer.subscribe(getTopics());
        return consumer;
    }


    private List<String> getTopics() {
        List<String> topics = new ArrayList<>();
        topics.add("message");
        return topics;
    }


}
